package com.dark.count;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by tengxue on 16-6-29.
 */
public class SentenceSpout extends BaseRichSpout{

    private ConcurrentHashMap<UUID,Values> pending;
    private SpoutOutputCollector collector;
    private String[] sentences={"my dog has fleas","i like cold beverages","the dog ate my homework","don't have a cow man","i don't think i like fleas"};

    private int index=0;
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("sentence"));
    }

    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector=spoutOutputCollector;
        pending=new ConcurrentHashMap<UUID, Values>();
    }

    /**
     * 使用了可靠的方式
     */
    public void nextTuple() {
        Values values=new Values(sentences[index]);
        UUID msgId=UUID.randomUUID();
        pending.put(msgId,values);
        collector.emit(values,msgId);
        index++;
        if (index>=sentences.length){
            index=0;
        }


    }

    /**
     * 成功处理了tuple
     * @param msgId
     */
    @Override
    public void ack(Object msgId) {
        pending.remove(msgId);
    }

    /**
     * 处理失败,重新发送数据
     * @param msgId
     */
    @Override
    public void fail(Object msgId) {
        collector.emit(pending.get(msgId),msgId);
    }
}
